package org.budo.elasticsearch.jdbc.driver;

import java.net.InetAddress;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import org.budo.elasticsearch.jdbc.driver.parser.jsql.ElasticsearchJSqlStatementParser;
import org.budo.jdbc.driver.AbstractBudoJdbcDriver;
import org.budo.jdbc.driver.parser.SqlStatementParser;
import org.budo.jdbc.driver.parser.cache.CachedJdbcStatementParser;
import org.budo.support.java.net.util.JavaNetUtil;
import org.budo.support.lang.util.MapUtil;
import org.budo.support.servlet.util.QueryStringUtil;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.settings.Settings.Builder;
import org.elasticsearch.common.transport.InetSocketTransportAddress;

import lombok.extern.slf4j.Slf4j;

/**
 * @author lmw
 */
@Slf4j
public class BudoElasticsearchJdbcDriver extends AbstractBudoJdbcDriver {
    private static final int DEFAULT_PORT = 9300;

    private static final String URL_PREFIX_1 = "jdbc:elasticsearch://";

    private static final String URL_PREFIX_2 = "jdbc:es://";

    private static final SqlStatementParser STATEMENT_PARSER = new CachedJdbcStatementParser() //
            .statementParser(new ElasticsearchJSqlStatementParser());

    static {
        try {
            DriverManager.registerDriver(new BudoElasticsearchJdbcDriver()); // 类加载时注册驱动
        } catch (Throwable e) {
            throw new IllegalStateException("#36 Can't register driver!", e);
        }
    }

    @Override
    public boolean acceptsURL(String url) throws SQLException {
        if (null == url || url.isEmpty()) {
            return false;
        }

        url = url.toLowerCase();
        return url.startsWith(URL_PREFIX_1) || url.startsWith(URL_PREFIX_2);
    }

    @SuppressWarnings("resource")
    @Override
    public Connection connect(String url, Properties info) throws SQLException {
        if (!this.acceptsURL(url)) {
            log.error("#53 not acceptsURL, url=" + url + ", info=" + info);
            return null;
        }

        int hostStart = url.startsWith(URL_PREFIX_1) ? URL_PREFIX_1.length() : URL_PREFIX_2.length(); // host start
        int hostEnd = url.indexOf(':', hostStart);
        if (hostEnd < 0) { // 未指定端口
            hostEnd = url.indexOf('/', hostStart);
        }

        if (hostEnd < 0) { // 未指定库名
            hostEnd = url.length();
        }

        String host = url.substring(hostStart, hostEnd);
        Integer port;

        int portEnd = url.indexOf('/', hostEnd);
        if (portEnd < 0) {
            port = DEFAULT_PORT;
        } else {
            port = Integer.parseInt(url.substring(hostEnd + 1, portEnd));
        }

        int indexStart = Math.max(portEnd, hostEnd);
        int parameterStart = url.indexOf('?');

        Map<String, String[]> parameters;
        String index;

        if (parameterStart < 0) { // 未指定参数
            index = url.substring(indexStart + 1);
            parameters = new HashMap<String, String[]>();
        } else {
            index = url.substring(indexStart + 1, parameterStart);
            parameters = QueryStringUtil.queryStringToMap(url.substring(parameterStart));
        }

        TransportClient transportClient = this.transportClient(host, port, parameters);

        return new BudoElasticsearchJdbcConnection() //
                .transportClient(transportClient) //
                .url(url) //
                .statementParser(STATEMENT_PARSER) //
                .catalog(index);
    }

    private TransportClient transportClient(String host, Integer port, Map<String, String[]> parameters) {
        Builder builder = Settings.builder();

        for (ESConf conf : ESConf.values()) {
            String value = MapUtil.getFirst(parameters, conf.getKey(), conf.getDefaultValue());
            builder.put(conf.getKey(), value);
        }

        Settings settings = builder.build();

        TransportClient transportClient = TransportClient.builder() //
                .settings(settings) //
                .build();

        InetAddress inetAddress = JavaNetUtil.getInetAddressByName(host);
        InetSocketTransportAddress transportAddress = new InetSocketTransportAddress(inetAddress, port);

        transportClient.addTransportAddress(transportAddress);
        return transportClient;
    }
}